package com.seamtop.cuber.core;

import com.seamtop.cuber.common.base.DataObject;
import com.seamtop.cuber.common.entriy.CuberReducerJob;
import com.seamtop.cuber.common.entriy.CuberReducer;
import com.seamtop.cuber.common.entriy.Duration;
import com.seamtop.cuber.common.entriy.TransData;
import com.seamtop.cuber.common.metadata.Column;
import com.seamtop.cuber.common.metadata.TableMetaData;
import com.seamtop.cuber.common.util.DateUtil;
import com.seamtop.cuber.common.util.StringUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.HashMap;

/**
 * Created by zongjunfeng on 2015/9/6.
 */
public class HqlAppendUtil {

    private static final Logger LOG = LoggerFactory.getLogger(HqlAppendUtil.class);


    public static String appendHql(CuberReducer reducer,Duration duration){
        String statType = reducer.getStatType();
        String statColumn = reducer.getStatColumn();
        String groupBy = getColumnStr(reducer.getColumnArray());
        String selectSql = appendSelectSql(statType, statColumn,groupBy);
        String fromSql = appendFromSql(reducer.getSourceTable());
        String whereSql = appendWhereSql(reducer,duration);
        String groupBySql = appendGroupBySql(groupBy);
        String hiveSql =  selectSql + fromSql + whereSql + groupBySql;
        return hiveSql;
    }


    private static String appendSelectSql(String statType,String statColumn,String groupBy){
        String statTypeSql = "";
        if("sum".equals(statType)){
            statTypeSql = "if(sum("+statColumn+") is not NULL,sum("+statColumn+"),0)";
        }else if("count".equals(statType)){
            if(StringUtil.isEmpty(statColumn) || "-1".equals(statColumn)){
                statTypeSql = "count(1)";
            }else{
                statTypeSql = "count(distinct "+statColumn+")";
            }
        }else if("avg".equals(statType)){
            statTypeSql = "if(avg("+statColumn+") is not NULL,avg("+statColumn+"),0)";
        }

        String selectSql = "select ";
        if(!StringUtil.isEmpty(groupBy)){
            selectSql += groupBy + ",";
        }
        selectSql += statTypeSql;
        return selectSql;
    }


    private static String appendFromSql(String sourceTable){
        String statTable = "HIVE_" + sourceTable;
        return " from " + statTable;
    }

    private static String appendWhereSql(CuberReducer reducer,Duration duration){
        String where = reducer.getWhere();
        String sourceTable = reducer.getSourceTable();
        TableMetaData metaData = DataObject.getMetaDataMap().get(sourceTable);
        String filterDateColumn = null;
        if(metaData.isVirtualTable()){
            filterDateColumn = "key";
        }else{
            filterDateColumn = reducer.getFilterDateColumn();
        }

        String keyPrefix = metaData.getTablePrefix();
        String startDateStr = DateUtil.getDurationDateStrOfDay(duration.getStartTime());
        String endDateStr = DateUtil.getDurationDateStrOfDay(duration.getEndTime());
        String whereSql = " where 1=1 ";
        if(!StringUtil.isEmpty(where)){
            whereSql += where;
        }
        if(!StringUtil.isEmpty(filterDateColumn)){
            if("key".equals(filterDateColumn)){
                whereSql +=   " and key >= '" + keyPrefix + "_"+ startDateStr + "' and key < '" + keyPrefix + "_" + endDateStr+"'";
            }else{
                whereSql +=  " and  " + filterDateColumn + " < " + duration.getEndTime() + " and " + filterDateColumn + ">"+ duration.getStartTime();
            }
        }
        return whereSql;
    }

    private static String appendGroupBySql(String groupBy){
        if(!StringUtil.isEmpty(groupBy)){
            return " group by " + groupBy;
        }else{
            return "";
        }
    }



    public static String getColumnStr(String [] columnArray){
        if(columnArray == null || columnArray.length == 0){
            return "";
        }
        String str = "";
        for(int i=0;i<columnArray.length;i++){
            str += columnArray[i];
            if(i != columnArray.length-1){
                str += ",";
            }
        }
        return str;
    }

    public static TransData combineResult(ResultSet rs,CuberReducerJob job) throws Exception{
        CuberReducer reducer = job.getReducer();
        String sourceTable = reducer.getSourceTable();
        String keyPrefix =  reducer.getReducerId();
        String [] columnArray = reducer.getColumnArray();
        TransData transData = new TransData();
        String rowKey = keyPrefix+ "_";
        String statPeriod = reducer.getStatPeriod();
        if(statPeriod.startsWith("recenty")){
            rowKey += statPeriod;
        }else{
            rowKey += DateUtil.combineBaseKeyOfDay(job.getDuration());
        }

        int count = 0;
        if(columnArray != null && columnArray.length>0){
            for(int i=0;i<columnArray.length;i++){
                String columnName = columnArray[i];
                if(StringUtil.isEmpty(columnName)){
                    continue;
                }

                TableMetaData tableMetaData = DataObject.getMetaDataMap().get(sourceTable);
                HashMap<String,Column> columnMap = tableMetaData.getColumnMap();
                Column column = columnMap.get(columnName);
                if(count == 0){
                    rowKey+= "|";
                }else{
                    rowKey+= "_";
                }
                rowKey+= column.getMatch();
                rowKey+= ":" + rs.getString(i+1);
            }
        }
        transData.setFamilyName("mf");
        transData.setRowKey(rowKey);
        transData.setColumnName("v");
        int length = 0;
        if(columnArray != null){
            length = columnArray.length;
        }
        transData.setValue(rs.getString(length + 1));
        return transData;
    }


    public static void dropTable(String tableName) throws Exception{
        Connection connection = null;
        Statement stmt = null;
        try{
            connection = HiveManager.getConnetion();
            stmt = connection.createStatement();
            stmt.executeQuery("DROP TABLE IF EXISTS " + tableName);
            LOG.info("Drop Hive Table["+tableName+"] success!");
        }catch (Exception e){
            LOG.error("Drop Hive Table[" + tableName + "] failed!", e);
            e.printStackTrace();
        }finally {
            stmt.close();
            connection.close();
        }
    }


    public static void executeSql(String sql) throws Exception{
        Connection connection = null;
        Statement stmt = null;
        try{
            connection = HiveManager.getConnetion();
            stmt = connection.createStatement();
            stmt.execute(sql);
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            stmt.close();
            connection.close();
        }
    }

    public static void executeQuerySql(String sql) throws Exception{
        Connection connection = null;
        Statement stmt = null;
        try{
            connection = HiveManager.getConnetion();
            stmt = connection.createStatement();
            stmt.execute(sql);
            ResultSet rs = stmt.executeQuery(sql);
            while(rs.next()){
                System.out.println(rs.getString(0));
            }
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            stmt.close();
            connection.close();
        }
    }


    public static void main(String [] args){
        try{
            dropTable("hive_stat_car_hit");
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}
